package com.ysb.hbase.dml;

import com.slimbloody.infra.warmup.WarmUp;
import com.ysb.hbase.model.*;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.MD5Hash;

import java.io.IOException;
import java.lang.reflect.Constructor;
import java.lang.reflect.Field;
import java.lang.reflect.InvocationTargetException;
import java.nio.charset.Charset;
import java.util.*;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;


@Slf4j
public class HbaseDmlService implements DmlService, WarmUp {
    private Configuration configuration;
    private volatile Connection connection;

    public void setConfiguration(Configuration configuration) {
        this.configuration = configuration;
    }

    public HbaseDmlService() {
    }

    public HbaseDmlService(Configuration configuration, Connection connection) {
        this.setConfiguration(configuration);
        this.setConnection(connection);
    }

    public void setConnection(Connection connection) {
        this.connection = connection;
    }

    public Connection getConnection() {
        if (null == this.connection) {
            synchronized (this) {
                if (null == this.connection) {
                    try {
                        ThreadPoolExecutor poolExecutor = new ThreadPoolExecutor(50, 200, 60L,
                                TimeUnit.SECONDS, new SynchronousQueue<Runnable>());
                        // init pool
                        poolExecutor.prestartCoreThread();
                        poolExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
                        this.connection = ConnectionFactory.createConnection(configuration, poolExecutor);
                    } catch (Exception e) {
                        log.error("hbase connection资源池创建失败", e);
                    }
                }
            }
        }
        return this.connection;
    }

    /**
     * 根据rowKey获得一条数据，返回 ValueModel
     * <p>
     * tableName, rowKey, columnFamily, qualifier 都必填
     *
     * @param getParam getParam
     * @return ValueModel
     */
    @Override
    public ValueModel getValue(HbaseGetParam getParam) {
        return getValue(getParam, false);
    }


    @Override
    public ValueModel getValueMd5(HbaseGetParam getParam) {
        return getValue(getParam, true);
    }


    private ValueModel getValue(HbaseGetParam getParam, boolean md5) {
        if (getParam == null || StringUtils.isBlank(getParam.getTableName())
                || StringUtils.isBlank(getParam.getRowKey())
                || StringUtils.isBlank(getParam.getColumnFamily())
                || StringUtils.isBlank(getParam.getQualifier())
        ) {
            throw new IllegalArgumentException("HbaseGetParam error：tableName, rowKey, columnFamily, qualifier must not be null");
        }

        HbaseGetRowParam getRowParam = HbaseGetRowParam.builder()
                .tableName(getParam.getTableName())
                .columnFamily(getParam.getColumnFamily())
                .rowKey(getParam.getRowKey())
                .qualifierList(Collections.singletonList(getParam.getQualifier()))
                .build();

        List<ValueModel> valueList = getValueList(getRowParam, md5);
        if (CollectionUtils.isEmpty(valueList)) {
            return new ValueModel();
        }
        return valueList.get(0);
    }


    /**
     * 根据rowKey获得一条数据，返回具体的value值
     * <p>
     * tableName, rowKey, columnFamily, qualifier 都必填
     *
     * @param getParam getParam
     * @return ValueModel
     */
    @Override
    public byte[] getValueByte(HbaseGetParam getParam) {
        ValueModel model = getValue(getParam, false);
        return model.getValue();
    }

    @Override
    public byte[] getValueByteMd5(HbaseGetParam getParam) {
        ValueModel model = getValue(getParam, true);
        return model.getValue();
    }


    /**
     * 根据rowKey获得数据列表，返回 List<ValueModel>
     * <p>
     * tableName, rowKey, 必填
     * columnFamily, qualifier 选填
     *
     * @param getParam getParam
     * @return HbaseGetResult
     */
    @Override
    public List<ValueModel> getValueList(HbaseGetRowParam getParam) {
        return getValueList(getParam, false);
    }

    @Override
    public List<ValueModel> getValueListMd5(HbaseGetRowParam getParam) {
        return getValueList(getParam, true);
    }

    private List<ValueModel> getValueList(HbaseGetRowParam getParam, boolean md5) {
        if (getParam == null || StringUtils.isBlank(getParam.getTableName())
                || StringUtils.isBlank(getParam.getColumnFamily())
                || StringUtils.isBlank(getParam.getRowKey())) {
            throw new IllegalArgumentException("HbaseGetParam error：tableName, columnFamily, rowKey must not be null");
        }

        String tableName = getParam.getTableName();
        String rowKey = getParam.getRowKey();
        String columnFamily = getParam.getColumnFamily();

        // 1. 封装get
        if (md5) {
            rowKey = MD5Hash.getMD5AsHex(Bytes.toBytes(rowKey));
        }
        Get get = new Get(Bytes.toBytes(rowKey));
        if (StringUtils.isNotBlank(columnFamily)) {
            get.addFamily(Bytes.toBytes(columnFamily));

            List<String> qualifierList = getParam.getQualifierList();
            if (CollectionUtils.isNotEmpty(qualifierList)) {
                for (String qua : qualifierList) {
                    if (StringUtils.isNotBlank(qua)) {
                        get.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(qua));
                    }
                }
            }
        }

        // 2. 读取数据
        try (Table table = this.getConnection().getTable(TableName.valueOf(tableName))) {
            Result result = table.get(get);

            // 3. 处理结果
            return getRowModel(result).getValueList();
        } catch (IOException e) {
            log.error("hbase getValueList error", e);
            return new ArrayList<>();
        }
    }

    @SuppressWarnings("unchecked")
    @Override
    public <T> T getRowObj(HbaseGetRowParam getParam, Class<T> t) throws Exception {
        List<ValueModel> valueList = getValueList(getParam);
        return (T) getTObject0(t, valueList);
    }

    @SuppressWarnings("unchecked")
    @Override
    public <T> T getRowObjMd5(HbaseGetRowParam getParam, Class<T> t) throws Exception {
        List<ValueModel> valueList = getValueListMd5(getParam);
        return (T) getTObject0(t, valueList);
    }

    private static <T> Object getTObject0(Class<T> t, List<ValueModel> valueList)
            throws NoSuchMethodException, InstantiationException, IllegalAccessException, InvocationTargetException {
        Constructor<?> constructor = t.getDeclaredConstructor();
        Object obj = constructor.newInstance();
        for (ValueModel valueModel : valueList) {
            String qualifier = valueModel.getQualifier();
            try {
                Field field = t.getDeclaredField(qualifier);
                Class<?> type = field.getType();
                if (type.isArray() && byte.class.equals(type.getComponentType())) {
                    field.setAccessible(true);
                    field.set(obj, valueModel.getValue());
                }
            } catch (Exception ignore) {
            }
        }
        return obj;
    }


    /**
     * 根据结果处理数据
     *
     * @param result result
     * @return RowModel
     */
    private static RowModel getRowModel(Result result) {
        Cell[] cells = result.rawCells();

        List<ValueModel> valueModels = new ArrayList<>();
        String rowKey = "";
        String cf0 = "";
        for (Cell cell : cells) {
            // 1. 获取rowKey
            byte[] rowByte = CellUtil.cloneRow(cell);
            if (StringUtils.isBlank(rowKey)) {
                rowKey = Bytes.toString(rowByte);
            }

            // 2. 获取 family
            String cfTemp = Bytes.toString(CellUtil.cloneFamily(cell));
            if (StringUtils.isBlank(cf0)) {
                cf0 = cfTemp;
            } else if (!cf0.equals(cfTemp)) {
                throw new RuntimeException("getRowResult: Multiple columnFamily exception");
            }

            // 3. 获取 qualifier
            String qualifier = Bytes.toString(CellUtil.cloneQualifier(cell));

            // 4. 获取value
            byte[] bytes = CellUtil.cloneValue(cell);

            // 5. 封装model
            ValueModel vm = ValueModel.builder()
                    .qualifier(qualifier)
                    .value(bytes)
                    .build();
            valueModels.add(vm);
        }
        return RowModel.builder().rowKey(rowKey).valueList(valueModels).build();
    }


    /**
     * 根据多个 rowKey获取数据，返回 list
     * tableName，columnFamily，rowKeyList 必填。
     * <p>
     * 这里在获取数据的时候，会转换成md5
     *
     * @param param param
     * @return List<RowModel>
     */
    @Override
    public List<RowModel> batchGet(HbaseBatchGetParam param) {
        if (param == null || StringUtils.isBlank(param.getTableName())
                || StringUtils.isBlank(param.getColumnFamily())
                || CollectionUtils.isEmpty(param.getRowKeyList())) {
            throw new IllegalArgumentException("HbaseBatchGetParam error：tableName, columnFamily, rowKeyList must not be null");
        }

        String tableName = param.getTableName();
        List<String> rowKeyList = param.getRowKeyList();
        String columnFamily = param.getColumnFamily();
        List<String> qualifierList = param.getQualifierList();

        // 1. 封装get
        List<Get> gets = new ArrayList<>();
        for (String key : rowKeyList) {
            if (StringUtils.isBlank(key)) {
                throw new IllegalArgumentException("HbaseBatchGetParam.rowKeyList error：rowKey must not be blank");
            }

            byte[] keyBytes = Bytes.toBytes(key);
            byte[] cfBytes = Bytes.toBytes(columnFamily);
            if (CollectionUtils.isNotEmpty(qualifierList)) {
                for (String qua : qualifierList) {
                    gets.add(new Get(keyBytes).addColumn(cfBytes, Bytes.toBytes(qua)));
                }
            } else {
                gets.add(new Get(keyBytes).addFamily(cfBytes));
            }
        }

        // 2. 读取数据
        try (Table table = this.getConnection().getTable(TableName.valueOf(tableName))) {
            Result[] results = table.get(gets);

            // 3. 处理结果
            List<RowModel> rowModels = new ArrayList<>();
            if (results != null) {
                for (Result result : results) {
                    RowModel rowModel = getRowModel(result);
                    if (StringUtils.isNotBlank(rowModel.getRowKey())) {
                        rowModels.add(rowModel);
                    }
                }
            }
            return rowModels;
        } catch (Exception e) {
            log.error("hbase batchGet error", e);
            return new ArrayList<>();
        }
    }

    @Override
    public List<RowModel> batchGetMd5(HbaseBatchGetParam param) {
        if (param == null || StringUtils.isBlank(param.getTableName())
                || StringUtils.isBlank(param.getColumnFamily())
                || CollectionUtils.isEmpty(param.getRowKeyList())) {
            throw new IllegalArgumentException("HbaseBatchGetParam error：tableName, columnFamily, rowKeyList must not be null");
        }

        String tableName = param.getTableName();
        List<String> rowKeyList = param.getRowKeyList();
        String columnFamily = param.getColumnFamily();
        List<String> qualifierList = param.getQualifierList();

        // 1. 封装get
        Map<String, String> rowKeyMap = new HashMap<>();
        List<Get> gets = new ArrayList<>();
        for (String key : rowKeyList) {
            if (StringUtils.isBlank(key)) {
                throw new IllegalArgumentException("HbaseBatchGetParam.rowKeyList error：rowKey must not be blank");
            }
            String md5Key = MD5Hash.getMD5AsHex(Bytes.toBytes(key));
            rowKeyMap.put(md5Key, key);
            byte[] keyBytes = Bytes.toBytes(md5Key);
            byte[] cfBytes = Bytes.toBytes(columnFamily);
            if (CollectionUtils.isNotEmpty(qualifierList)) {
                for (String qua : qualifierList) {
                    gets.add(new Get(keyBytes).addColumn(cfBytes, Bytes.toBytes(qua)));
                }
            } else {
                gets.add(new Get(keyBytes).addFamily(cfBytes));
            }
        }

        // 2. 读取数据
        try (Table table = this.getConnection().getTable(TableName.valueOf(tableName))) {
            Result[] results = table.get(gets);

            // 3. 处理结果
            List<RowModel> rowModels = new ArrayList<>();
            if (results != null) {
                for (Result result : results) {
                    RowModel rowModel = getRowModel(result);
                    if (StringUtils.isNotBlank(rowModel.getRowKey())) {
                        rowModels.add(rowModel);
                    }
                }
            }
            rowModels.forEach(a -> a.setRowKey(rowKeyMap.get(a.getRowKey())));
            return rowModels;
        } catch (Exception e) {
            log.error("hbase batchGet error", e);
            return new ArrayList<>();
        }
    }


    /**
     * 根据多个 rowKey 获取数据，返回 map:<rowKey, List<ValueModel>>
     * tableName，columnFamily，rowKeyList 必填。
     *
     * @param getParam getParam
     * @return List<RowModel>
     */
    @Override
    public Map<String, List<ValueModel>> batchGetMap(HbaseBatchGetParam getParam) {
        List<RowModel> rowModels = batchGet(getParam);
        return rowModels.stream().collect(Collectors.toMap(RowModel::getRowKey, RowModel::getValueList, (a, b) -> a));
    }

    @Override
    public Map<String, List<ValueModel>> batchGetMapMd5(HbaseBatchGetParam getParam) {
        List<RowModel> rowModels = batchGetMd5(getParam);
        return rowModels.stream().collect(Collectors.toMap(RowModel::getRowKey, RowModel::getValueList, (a, b) -> a));
    }

    /**
     * 根据多个 rowKey 获取数据，返回 map:<rowKey, T>
     * tableName，columnFamily，rowKeyList 必填。
     *
     * @param getParam getParam
     * @return List<RowModel>
     */
    @Override
    @SuppressWarnings("unchecked")
    public <T> Map<String, T> batchGetMap(HbaseBatchGetParam getParam, Class<T> t) {
        List<RowModel> rowModels = batchGet(getParam);
        return rowModels.stream()
                .collect(Collectors.toMap(RowModel::getRowKey, a -> {
                    try {
                        return (T) getTObject0(t, a.getValueList());
                    } catch (Exception e) {
                        throw new RuntimeException(e);
                    }
                }, (a, b) -> a));
    }

    @Override
    @SuppressWarnings("unchecked")
    public <T> Map<String, T> batchGetMapMd5(HbaseBatchGetParam getParam, Class<T> t) {
        List<RowModel> rowModels = batchGetMd5(getParam);
        return rowModels.stream()
                .collect(Collectors.toMap(RowModel::getRowKey, a -> {
                    try {
                        return (T) getTObject0(t, a.getValueList());
                    } catch (Exception e) {
                        throw new RuntimeException(e);
                    }
                }, (a, b) -> a));
    }

    @Override
    public boolean putData(HbasePutParam putParam) {
        return putData(putParam, false);
    }

    @Override
    public boolean putDataMd5(HbasePutParam putParam) {
        return putData(putParam, true);
    }

    private boolean putData(HbasePutParam putParam, boolean md5) {
        if (putParam == null || StringUtils.isBlank(putParam.getTableName())
                || StringUtils.isBlank(putParam.getRowKey())
                || StringUtils.isBlank(putParam.getColumnFamily())
                || CollectionUtils.isEmpty(putParam.getValueList())
        ) {
            throw new IllegalArgumentException("HbasePutParam error：tableName, rowKey, columnFamily, valueList must not be null");
        }
        List<ValueModel> valueList = putParam.getValueList();
        if (valueList.stream().anyMatch(a -> StringUtils.isEmpty(a.getQualifier()))) {
            throw new IllegalArgumentException("HbasePutParam.ValueList error：qualifier must not be null");
        }

        String tableName = putParam.getTableName();
        String rowKey = putParam.getRowKey();
        String columnFamily = putParam.getColumnFamily();
        // 1. 封装put
        RowModel rowModel = RowModel.builder().valueList(valueList).rowKey(rowKey).build();
        Put put = getPut(rowModel, columnFamily, md5);

        // 2. 发起请求
        try (Table table = this.getConnection().getTable(TableName.valueOf(tableName))) {
            table.put(put);
            return true;
        } catch (Exception e) {
            log.error("hbase putData error", e);
            return false;
        }
    }

    @Override
    public boolean batchPut(HbaseBatchPutParam param) {
        return batchPut(param, false);
    }

    @Override
    public boolean batchPutMd5(HbaseBatchPutParam param) {
        return batchPut(param, true);
    }

    private boolean batchPut(HbaseBatchPutParam param, boolean md5) {
        if (param == null || StringUtils.isBlank(param.getTableName())
                || StringUtils.isBlank(param.getColumnFamily())
                || CollectionUtils.isEmpty(param.getRowModels())
        ) {
            throw new IllegalArgumentException("HbaseBatchPutParam error：tableName, columnFamily, rowList must not be null");
        }

        List<RowModel> rowList = param.getRowModels();
        if (rowList.stream().anyMatch(a -> StringUtils.isBlank(a.getRowKey())
                || CollectionUtils.isEmpty(a.getValueList())
                || a.getValueList().stream().anyMatch(b -> StringUtils.isBlank(b.getQualifier()))
        )) {
            throw new IllegalArgumentException("HbaseBatchPutParam.rowList error：rowKey, valueList, valueList.qualifier must not be null");
        }

        String tableName = param.getTableName();
        String columnFamily = param.getColumnFamily();

        // 1. 封装put
        List<Put> puts = new ArrayList<>();
        for (RowModel rowModel : rowList) {
            puts.add(getPut(rowModel, columnFamily, md5));
        }

        // 2. 发起请求
        try (Table table = this.getConnection().getTable(TableName.valueOf(tableName))) {
            table.put(puts);
            return true;
        } catch (Exception e) {
            log.error("hbase batchPut error", e);
            return false;
        }
    }

    private static Put getPut(RowModel rowModel, String columnFamily, boolean md5) {
        String rowKey = rowModel.getRowKey();
        if (md5) {
            rowKey = MD5Hash.getMD5AsHex(Bytes.toBytes(rowKey));
        }
        Put put = new Put(rowKey.getBytes()); // 行键

        List<ValueModel> valueList = rowModel.getValueList();
        for (ValueModel valueModel : valueList) {
            String qualifier = valueModel.getQualifier();
            byte[] value = valueModel.getValue();
            put.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(qualifier), value); // 列族、限定符、值
        }
        return put;
    }

    public static final String rowKey = "123";

    @Override
    public boolean warm() {
        try {
            connection.getTable(TableName.valueOf("123")).get(new Get(rowKey.getBytes(Charset.defaultCharset())));
        } catch (Exception e) {
            log.warn("prehot: cant get value", e);
        }
        return false;
    }
}
